package cn.com.ytst.data.integration.core.channel;

import java.util.Collection;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;

/**
 * @ClassName DefaultChannel
 * @Deacription
 * @Author Administrator
 * @Date 2020/1/18 0018 17:24
 * @Version 1.0
 **/
@Slf4j
public class MemoryPipleChannel implements PipleChannel {

    //private Logger logger = LoggerFactory.getLogger(MemoryPipleChannel.class);

    private LinkedBlockingQueue<String> queue = null;

    private AtomicLong memoryBytes = null;

    private ReentrantLock lock = new ReentrantLock();
    private Condition empty ;//写锁

    private Condition notEmpty ;//读锁

    private int queuesize = 10000;

    private int initcount = 0;

    public MemoryPipleChannel() {

        queue = new LinkedBlockingQueue<>(queuesize);
        memoryBytes = new AtomicLong(initcount);
        empty = lock.newCondition();//写锁
        notEmpty = lock.newCondition();//读锁
    }

    //往里面放入数据
    @Override
    public void doPush(String record) {
    	final ReentrantLock lock = this.lock;
       

      
        try {
        	 lock.lockInterruptibly();
        	while (size() == queuesize ) {
            	
            	//说明队列满了
                empty.await();
                
            }
            notEmpty.signal();
            //如果不够所定义的queuesize大小了，notEmpty会一直被阻塞着
            long length = memoryBytes.get() + record.getBytes("utf-8").length;
            memoryBytes.set(length);
            queue.put(record);

        } catch (Exception e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }


    }

    //批量插入数据
    @Override
    public void doPushAll(Collection<String> records) {

    }




    //拉取数据
    @Override
    public String doPull() {
    	final ReentrantLock lock = this.lock;
    	try {
			lock.lockInterruptibly();
		} catch (InterruptedException e1) {
			 Thread.currentThread().interrupt();
		}
        while (size() == 0) {
        	
            try {
            	notEmpty.await();
            	//break;
            } catch (Exception e) {
                Thread.currentThread().interrupt();
            }
        }
        try {
        	String line = this.queue.take();
        	empty.signal();
            //阻塞写线程，让读线程进行读操作
//            if(this.queue.isEmpty()){
//                return null;
//            }
            return line;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }finally{
            lock.unlock();
        }
    }

    @Override
    public boolean isEmpty() {
        return this.queue.isEmpty();
    }

    @Override
    public int size() {
        return this.queue.size();
    }

    @Override
    public void readflush() {
        notEmpty.signal();
    }


}
